package person.tavion;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.junit.jupiter.api.Test;
import person.tavion.service.ConsumerService;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.Properties;

public class ConsumerTest2 implements ConsumerService {

    @Test
    @Override
    public void Consumer() {
        //
        Properties pro = KafkaUtil.getProperties(TypeEnum.Consumer);
        pro.setProperty("client.id", "ConsumerTest2");
        Consumer<String, String> consumer = new KafkaConsumer<>(pro);
        // 订阅主题
        consumer.subscribe(Collections.singletonList(KafkaUtil.topicName));
        System.out.println("ConsumerTest2 - 开启成功");
        while(true){

            try {
                // 消费消息
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

                for (ConsumerRecord<String, String> record : records) {
                    if (record.key().startsWith("sendFail")) {

                        throw new RuntimeException("发送失败 fail");

                    }
                    System.out.printf(LocalDateTime.now() + " offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                    consumer.commitSync();
                }
            }catch (Exception e){
                e.printStackTrace();
            }


//        System.exit(0);

//            return;
        }
//        consumer.close();
    }

    @Override
    public void list() {

    }
}
